package com.lizhiyu.flink.monitor;

import com.lizhiyu.flink.model.AccessLogDO;
import com.lizhiyu.flink.model.TimeUtils;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;

/**
 * @Author:lizhiyu
 * @Date: 2022/2/13 14:12
 */
public class AccessLogSource  extends RichParallelSourceFunction<AccessLogDO> {
    private volatile Boolean flag = true;
    private Random random = new Random();
    //接口
    private static List<AccessLogDO> urlList = new ArrayList<>();
    static {
        urlList.add(new AccessLogDO("首页","/pub/api/v1/web/index_card","GET",200,"",new Date(),"",""));
        urlList.add(new AccessLogDO("个人信息","/pub/api/v1/web/user_info","GET",200,"",new Date(),"",""));
    }
    //状态码
    private static List<Integer> codeList = new ArrayList<>();
    static {
        codeList.add(200);
        codeList.add(200);
        codeList.add(502);
        codeList.add(403);
    }
    @Override
    public void run(SourceContext<AccessLogDO> ctx) throws Exception {
        while (flag){
            Thread.sleep(1000);
            int userId = random.nextInt(50);
            int httpCodeNum = random.nextInt(codeList.size());
            int accessLogNum = random.nextInt(urlList.size());
            AccessLogDO accessLogDO = urlList.get(accessLogNum);
            accessLogDO.setHttpCode(codeList.get(httpCodeNum));
            accessLogDO.setUserId(userId+"");
            //模拟迟到数据,100秒波动
            //long timestamp = System.currentTimeMillis() - random.nextInt(100000);
            long timestamp = System.currentTimeMillis() - random.nextInt(5000);
            accessLogDO.setCreateTime(new Date(timestamp));
            System.out.println("产生:"+accessLogDO.getTitle()+"，状态:"+accessLogDO.getHttpCode()+", 时间:"+ TimeUtils.format(accessLogDO.getCreateTime().getTime()));
            ctx.collect(accessLogDO);
        }
    }
    @Override
    public void cancel() {
        flag = false;
    }
}